package com.allwe.client.concurrent;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.concurrent.CountDownLatch;

/**
 * 生产者多线程安全 - 测试demo
 *
 * @Author: AllWe
 * @Date: 2024/09/27/9:30
 */
@Data
@Slf4j
public class ConcurrentProducer implements Runnable {

    /**
     * 消息体
     */
    private ProducerRecord<String, String> record;

    /**
     * 生产者
     */
    private KafkaProducer<String, String> producer;

    /**
     * 发令枪
     */
    private CountDownLatch countDownLatch;

    public ConcurrentProducer(ProducerRecord<String, String> record, KafkaProducer<String, String> producer, CountDownLatch countDownLatch) {
        this.record = record;
        this.producer = producer;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        try {
            String name = Thread.currentThread().getName();
            producer.send(record, new ConcurrentCallBackImpl(name));
            countDownLatch.countDown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
